package com.hy.study.mapper;

import com.hy.study.bean.SortFlowBean;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * 输入key 文件行 输入value 文件的一行数据 输出 key 是可以排序的bean对象 输出 value 是手机号
 *
 * @author huyong
 * @date 2024 年 02 月 05 日 15:12: 54
 */
public class SortFlowCountMapper extends Mapper<LongWritable, Text, SortFlowBean, Text> {
    private Text v = new Text();
    private SortFlowBean sortFlowBean = new SortFlowBean();

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, SortFlowBean, Text>.Context context) throws IOException, InterruptedException {
        String line = value.toString();
        String[] fileds = line.split("\t");
        int size = fileds.length;
        // 封装对象
        v.set(fileds[0]);
        String filed = fileds[1];
        if (StringUtils.isNotBlank(filed) && StringUtils.isNumeric(filed)) {
            sortFlowBean.setUpFlow(Long.parseLong(filed));
        } else {
            sortFlowBean.setUpFlow(0L);
        }
        filed = fileds[2];
        if (StringUtils.isNotBlank(filed) && StringUtils.isNumeric(filed)) {
            sortFlowBean.setDownFlow(Long.parseLong(filed));
        } else {
            sortFlowBean.setDownFlow(0L);
        }
        sortFlowBean.setSumFlow(sortFlowBean.getUpFlow() + sortFlowBean.getDownFlow());
        context.write(sortFlowBean, v);
    }
}
